package com.sbp.message.consumer;

import java.util.*;

import com.sbp.message.api.service.MessageSysInfoService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by wangmin on 2019/4/19.
 */
public class KafkaNormalConsumerThread extends KafkaBaseConsumerThread {
    private static final Logger logger = LoggerFactory.getLogger(KafkaNormalConsumerThread.class);

    private final MessageConsumer messageConsumer;
    private final String group;
    private final Set<String> topics;
    private final int threadIndex;
    private final KafkaConsumer<String, String> kafkaConsumer;

    public KafkaNormalConsumerThread(MessageConsumer messageConsumer, String group, Set<String> topics, int threadIndex,
            Properties kafkaProperties) {
        this.messageConsumer = messageConsumer;
        this.group = group;
        this.topics = topics;

        this.threadIndex = threadIndex;

        kafkaProperties.put("group.id", MessageSysInfoService.generateKafkaRealGroupId(group));
        this.kafkaConsumer = new KafkaConsumer<>(kafkaProperties);
    }

    @Override
    public void run() {
        try {
            Set<String> kafkaRealTopics = new HashSet<>();
            for (String topic : topics) {
                String kafkaRealTopicName = MessageSysInfoService.generateKafkaRealTopicName(topic);
                kafkaRealTopics.add(kafkaRealTopicName);
            }
            kafkaConsumer.subscribe(kafkaRealTopics);

            while (isRun) {
                try {
                    ConsumerRecords<String, String> records = kafkaConsumer.poll(10000);
                    if (records.isEmpty()) {
                        continue;
                    }
                    kafkaConsumer.commitSync();

                    for (ConsumerRecord<String, String> record : records) {
                        try {
                            messageConsumer.consumeMessage(record.value(), group);
                        } catch (Exception e) {
                            logger.warn("message = {}", record.value());
                            logger.warn("", e);
                        }
                    }
                } catch (Exception e) {
                    logger.error("", e);
                }
            }
        } finally {
            kafkaConsumer.close();
        }
    }

    @Override
    public String threadName() {
        return "KafkaNormalConsumer_" + group + "_" + threadIndex;
    }

    @Override
    public void shutdown() {
        kafkaConsumer.close();
        isRun = false;
    }

}

